/**
 * Copyright 2011 AJG van Schie
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */
package com.googlecode.julius.utils;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import julius.utilities.CollectionHelper;

public class ParallelForEach<T,U> {

	private final ExecutorService pool;
	private final ProgressCallback parent;
	private final boolean skipExceptions;
	private final int retry;
	private final List<Throwable> errors = CollectionHelper.list();
	

	public static interface ParallelCall<T,U>{
		T map(U u);
	}
	
	public static interface ProgressCallback{
		void progress(final int part, final int total);
	}
	
	private final class RetryCallable implements Callable<T> {
		private final ParallelCall<T, U> call;
		private final U u;

		private RetryCallable(final ParallelCall<T, U> call, final U u) {
			this.call = call;
			this.u = u;
		}

		@Override
		public T call() {
			int attempt = 0;
			RuntimeException t = null;
			do {
				try {
					return call.map(u);
				} catch (RuntimeException e) {
					attempt++;
					t = e;
				}
			}while(attempt <= retry);
			throw t;
		}
	}
	
	public ParallelForEach(final int  threadcount, final ProgressCallback parent ,final boolean skipExceptions, final int retry) {
		this(Executors.newFixedThreadPool(threadcount), parent, skipExceptions, retry);
	}
	
	public ParallelForEach(final ExecutorService pool, final ProgressCallback parent ,final boolean skipExceptions, final int retry) {
		this.pool = pool;
		this.parent = parent;
		this.skipExceptions = skipExceptions;
		this.retry = retry;
	}

	public Collection<T> map(final Collection<U> ori, final ParallelCall<T,U> call){
		errors.clear();
		List<T> items = CollectionHelper.list();
		try {
			// we have one call for reading the milestone, and x for all tickets
			parent.progress(1, ori.size() + 1);

			List<Future<T>> futures = CollectionHelper.list();

			// create futures
			for (final U u : ori) {
				Future<T> future = pool.submit(new RetryCallable(call, u));
				futures.add(future);
			}
			// call futures
			for (Future<T> future : futures) {
				T res = null;
				try {
					res = future.get();
					items.add(res);
				} catch (Throwable e) {
					errors.add(e);
				}
				parent.progress(1 + items.size(), ori.size() + 1);
			}
		} catch (Throwable e) {
			errors.add(e);
		}
		
		if(!skipExceptions && !errors.isEmpty()){
			throw new CombinedThrowable(errors);
		}
		
		return items;
	}
	
	public static class CombinedThrowable extends RuntimeException{
		
		private final Collection<Throwable> excs;
		
		public CombinedThrowable(final Collection<Throwable> excs){
			this.excs = excs;
		}
	}
	
	
}
